package com.ylkj.bus.onoffline.report;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.monitor.equipment.model.EquipmentAlarmMonitorLogEnty;
import com.ylkj.bus.onoffline.monitor.equipment.constant.DynamicsPropertyConstant;
import com.ylkj.bus.onoffline.monitor.equipment.constant.EquipmentAlarmConstant;
import com.ylkj.bus.onoffline.monitor.equipment.errorcode.EquipmentProxyEnum;
import com.ylkj.bus.onoffline.monitor.rule.code.RuleReportProxyEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @Author: wengzx
 * @Date: 2025/4/23
 * @Desc:
 */
@Slf4j
public class ReportOnlineStatusHandler extends ProcessFunction<MessageContent, MessageContent> {
    private final RedisConfig redisConfig;
    private final String targetKey;
    private Jedis jedis;
    private final OutputTag<EquipmentAlarmMonitorLogEnty> outTagStatus;

    public ReportOnlineStatusHandler(RedisConfig redisConfig,OutputTag<EquipmentAlarmMonitorLogEnty> outTagStatus) {
        this.redisConfig = redisConfig;
        this.outTagStatus = outTagStatus;
       targetKey = redisConfig.getTargetKey();
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = redisConfig.getJedis();
    }

    @Override
    public void processElement(MessageContent mqttMsg, ProcessFunction<MessageContent, MessageContent>.Context ctx, Collector<MessageContent> out) throws Exception {
        JSONObject body = JSONObject.parseObject(mqttMsg.getBody());
        if(body.isEmpty()){
            log.error("原始数据信息：{} 上报的数据为空," ,mqttMsg);
            return;
        }
        try {
            reportOnlineStatus(body,ctx);
        } catch (Exception e) {
            log.error("原始数据信息：{}", mqttMsg);
            log.error(e.getMessage(), e);
        }
    }

    private void reportOnlineStatus(JSONObject body, ProcessFunction<MessageContent, MessageContent>.Context ctx) {
        AlarmReportData reportAppStatus = body.to(AlarmReportData.class);
        Map<String, Object> map = new HashMap<>();
        AlarmReportData.Param param = reportAppStatus.getParam();
        String deviceCode = reportAppStatus.getDeviceId();
        String event = param.getEvent();
        String timestamp = reportAppStatus.getTimestamp();
        DateTime date = DateUtil.parse(timestamp);
        if(event.equals(RuleReportProxyEnum.DEVICE_ONLINE_PROXY.getEvent())){
            String hget = jedis.hget(targetKey, deviceCode);
            if(Objects.isNull(hget)){
                return;
            }else {
                // 获取出来的 hget 是 yyyy-MM-dd HH:mm:ss 类型的，要比较上报的值与 Redis 中 hget 如果小于，那么就 return
                DateTime redisDate = DateUtil.parse(hget);
                if (date.isBefore(redisDate)) { // 如果上报的时间小于 Redis 中存储的时间，则直接返回
                    return;
                }
            }
        }
        if(event.equals(RuleReportProxyEnum.DEVICE_OFFLINE_PROXY.getEvent())){
            jedis.hset(targetKey, deviceCode, timestamp);
        }
        //根据eventId获取需要发送的指标
        String reportProxy = RuleReportProxyEnum.getReportProxy(event);
        if (StrUtil.isBlank(reportProxy)) {
            log.info("告警数据为：【{}】的指标为空不予发送", reportAppStatus);
            return;
        }
        EquipmentAlarmMonitorLogEnty equipmentAlarmMonitorLogEnty = new EquipmentAlarmMonitorLogEnty();
        //设置设备标识
        equipmentAlarmMonitorLogEnty.setDeviceCode(deviceCode);
        //设置指标
        equipmentAlarmMonitorLogEnty.setEventType(reportProxy);
        String alarmCover;
        //获取到指标判断是恢复还是告警
        boolean recovery = RuleReportProxyEnum.isRecovery(event);
        if (recovery) {
            //恢复自动处理
            alarmCover = EquipmentAlarmConstant.ALARM_COVER_AUTO;
        }else {
            //设置告警时间
            equipmentAlarmMonitorLogEnty.setAlarmTime(date);
            equipmentAlarmMonitorLogEnty.setAlertLevel(EquipmentAlarmConstant.ALARM_LEVEL_CRITICAL);
            //告警未处理
            alarmCover = EquipmentAlarmConstant.ALARM_COVER_UNPROCESSED;
            //消缺工具
            String blankingTool = "DEFECT_ELIMINATION";
            map.put(DynamicsPropertyConstant.deviceOnlineStatus, event);
            equipmentAlarmMonitorLogEnty.setMessageSuggestion(EquipmentProxyEnum.LAST_OFFLINE.getProxySuggest());
            equipmentAlarmMonitorLogEnty.setAlertDescription(StrUtil.format(EquipmentProxyEnum.LAST_OFFLINE.getProxyDesc(), Objects.nonNull(param.getCurrent()) ? param.getCurrent() : StrUtil.SPACE));
            equipmentAlarmMonitorLogEnty.setAlertSource(EquipmentProxyEnum.LAST_OFFLINE.getProxyOrigin());
            equipmentAlarmMonitorLogEnty.setBlankingTool(blankingTool);
            // 将 map 转换为 JSON 字符串
            String mapAsJson = JSON.toJSONString(map);
            equipmentAlarmMonitorLogEnty.setEventValue(mapAsJson);
            equipmentAlarmMonitorLogEnty.setAlertDimensions("1");
            equipmentAlarmMonitorLogEnty.setNetCode(EquipmentAlarmConstant.NET_CODE_INNER);
        }
        equipmentAlarmMonitorLogEnty.setAlarmCover(alarmCover);
        equipmentAlarmMonitorLogEnty.setNetCode(EquipmentAlarmConstant.NET_CODE_INNER);
        if (recovery) {
            equipmentAlarmMonitorLogEnty.setPersonnelHandler("");
            equipmentAlarmMonitorLogEnty.setHandlingTime(DateUtil.date());
        }
        ctx.output(outTagStatus,equipmentAlarmMonitorLogEnty);
    }


    @Override
    public void close() throws Exception {
        if (jedis != null) {
            jedis.close();
        }
        super.close();
    }
}
